package com.blackcattle.data.migrate.mongo.service;

import cn.hutool.core.util.StrUtil;
import com.blackcattle.data.migrate.mongo.config.RuleConfig;
import com.blackcattle.data.migrate.mongo.config.SourceConfig;
import com.blackcattle.data.migrate.mongo.config.TargetConfig;
import com.blackcattle.data.migrate.mongo.config.entity.TransferRule;
import com.blackcattle.data.migrate.mongo.helper.MongoDBHelper;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.bson.BsonDocument;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author dougs@sina.cn
 * @version 1.0.0
 */
@Data
@Slf4j
@Service
@ConditionalOnProperty(value = "data.migrate.plug", havingValue = "mongo")
public class MongoMigrateService {
    @Autowired
    private SourceConfig sourceConfig;
    @Autowired
    private TargetConfig targetConfig;
    @Autowired
    private RuleConfig ruleConfig;

    private MongoDBHelper sourceDBHelper;
    private MongoDBHelper targetDBHelper;

    private SimpleDateFormat sdf = new SimpleDateFormat();

    /**
     * 最新增量值
     */
    private Map<String, Object> lastIncrementValue = new HashMap<>();

    @PostConstruct
    private void initDbHelper(){
        sourceDBHelper = new MongoDBHelper(sourceConfig.getHost(), sourceConfig.getPort(), sourceConfig.getAuthDb(),
                sourceConfig.getDatabase(), sourceConfig.getUsername(), sourceConfig.getPassword());

        targetDBHelper = new MongoDBHelper(targetConfig.getHost(), targetConfig.getPort(), targetConfig.getAuthDb(),
                targetConfig.getDatabase(), targetConfig.getUsername(), targetConfig.getPassword());

        if(StrUtil.isNotBlank(sourceConfig.getIncrementField())){
            lastIncrementValue.put(sourceConfig.getIncrementField(), sourceConfig.getIncrementInit());
        }

        log.info("Mongo数据库连接创建成功！");
    }

    @PreDestroy
    public void preDestroy() {
        if(sourceDBHelper != null){
            sourceDBHelper.close();
            sourceDBHelper = null;
        }
        if(targetDBHelper != null){
            targetDBHelper.close();
            targetDBHelper = null;
        }
        log.info("Mongo数据库连接释放成功！");
    }

    @Scheduled(fixedDelayString = "${data.migrate.fixedDelay:5000}")
    public void dataMigrate(){
        String command = sourceConfig.getFindCommand();
        String incrementReplace = sourceConfig.getIncrementReplace();
        if(StrUtil.isNotBlank(incrementReplace) && command.indexOf(incrementReplace)>0){
            command = command.replaceAll(incrementReplace, lastIncrementValue.get(sourceConfig.getIncrementField()).toString());
        }
        log.info("迁移数据，执行命令：{}", command);

        BsonDocument filter = BsonDocument.parse(command);
        BsonDocument sort = BsonDocument.parse(sourceConfig.getSortCommand());
        List<Map> list = sourceDBHelper.findList(sourceConfig.getCollection(), filter, sort, sourceConfig.getLimit());
        if(!CollectionUtils.isEmpty(list)){
            if(StrUtil.isNotBlank(sourceConfig.getIncrementField())){
                lastIncrementValue.put(sourceConfig.getIncrementField(), list.get(list.size()-1).get(sourceConfig.getIncrementField()));
            }

            Map<String, Integer> count = new HashMap<>();
            count.put("fault", 0);

            list.forEach(map -> {
                try {
                    this.transfer(map);
                    targetDBHelper.insert(targetConfig.getCollection(), map);
                }catch (Exception ex){
                    count.put("fault", count.get("fault")+1);
                    log.error("迁移数据，同步数据失败：{}", map.toString(), ex);
                }
            });
            log.info("迁移数据，同步成功数据{}条", list.size()-count.get("fault"));
        }else{
            log.info("迁移数据，同步完成，退出程序!");
            System.exit(0);
        }
    }

    /**
     * 根据规则转换数据
     * @author dougs@sina.cn
     * @date 2021/4/23 18:49
     * @param map
     * @return void
     */
    private void transfer(Map map){
        List<TransferRule> rules = ruleConfig.getRules();
        if(!CollectionUtils.isEmpty(rules)){
            rules.forEach(transferRule -> {
                String field = transferRule.getField();
                String action = transferRule.getAction();
                String value = transferRule.getValue();
                if("add".equalsIgnoreCase(action)){
                    map.put(field, value);
                }
                if("delete".equalsIgnoreCase(action)){
                    map.remove(field);
                }
                if("rename".equalsIgnoreCase(action)){
                    Object object = map.get(field);
                    if(object != null){
                        map.put(value, object);
                    }
                    map.remove(field);
                }
                if("copy".equalsIgnoreCase(action)){
                    Object object = map.get(field);
                    if(object != null){
                        map.put(value, object);
                    }
                }
                if("date".equalsIgnoreCase(action)){
                    Object object = map.get(field);
                    if(object != null){
                        sdf.applyPattern(transferRule.getValue());
                        try {
                            map.put(field, sdf.parse(object.toString()));
                        } catch (ParseException e) {
                            log.error("数据处理异常", e);
                        }
                    }
                }
                if("format".equalsIgnoreCase(action)){
                    Object object = map.get(field);
                    if(object != null){
                        sdf.applyPattern(transferRule.getValue());
                        map.put(field, sdf.format(object));
                    }
                }
                if("change".equalsIgnoreCase(action)){
                    Object object = map.get(field);
                    if(object != null){
                        if("Integer".equalsIgnoreCase(value)){
                            map.put(field, new Integer(object.toString()));
                        }
                        if("Long".equalsIgnoreCase(value)){
                            map.put(field, new Long(object.toString()));
                        }
                        if("Float".equalsIgnoreCase(value)){
                            map.put(field, new Float(object.toString()));
                        }
                        if("Double".equalsIgnoreCase(value)){
                            map.put(field, new Double(object.toString()));
                        }
                        if("BigDecimal".equalsIgnoreCase(value)){
                            map.put(field, new BigDecimal(object.toString()));
                        }
                    }
                }
                if("seconds".equalsIgnoreCase(action)){
                    Object object = map.get(field);
                    if(object != null && object instanceof Date){
                        map.put(value, ((Date)object).getTime()/1000);
                    }
                }
                if("milliseconds".equalsIgnoreCase(action)){
                    Object object = map.get(field);
                    if(object != null && object instanceof Date){
                        map.put(value, ((Date)object).getTime());
                    }
                }
            });
        }
    }

}
